package com.mini.rpc.provider.registry;

import com.alibaba.nacos.api.exception.NacosException;
import com.alibaba.nacos.api.exception.runtime.NacosSerializationException;
import com.alibaba.nacos.api.naming.NamingFactory;
import com.alibaba.nacos.api.naming.NamingService;
import com.alibaba.nacos.api.naming.pojo.Instance;
import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.mini.rpc.common.RpcServiceHelper;
import com.mini.rpc.common.ServiceMeta;
import com.mini.rpc.provider.registry.loadbalancer.LoadBalancerFactory;
import com.mini.rpc.provider.registry.loadbalancer.LoadBalancerType;
import com.mini.rpc.provider.registry.loadbalancer.ServiceLoadBalancer;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.util.*;
import java.util.stream.Collectors;

/**
 * Nacos 注册中心, 用配置文件来控制, 注册中心、负载均衡算法、序列化方式
 *
 * @author: hliushi
 * @date: 2022/4/23 8:10
 **/
@Slf4j
public class NacosRegistryService implements RegistryService {

    private static final Set<String> serviceNames = new HashSet<>();
    private static NamingService namingService = null;
    private static InetSocketAddress address;
    private static ObjectMapper mapper = new ObjectMapper();

    /**
     * @param registryAddr cvm-node02:8848
     */
    public NacosRegistryService(String registryAddr) throws Exception {
        namingService = getNamingService(registryAddr);
    }

    private NamingService getNamingService(String registryAddr) {
        try {
            return NamingFactory.createNamingService(registryAddr);
        } catch (NacosException e) {
            throw new RuntimeException("Error connecting to Nacos: ", e);
        }
    }

    @Override
    public void register(ServiceMeta serviceMeta) throws Exception {
        Instance instance = builderInstance(serviceMeta);
        namingService.registerInstance(
                RpcServiceHelper.buildServiceKeyWith_(serviceMeta.getServiceName(), serviceMeta.getServiceVersion()),
                instance);

        address = new InetSocketAddress(serviceMeta.getServiceAddr(), serviceMeta.getServicePort());
        serviceNames.add(serviceMeta.getServiceName());
    }

    @Override
    public void unRegister(ServiceMeta serviceMeta) throws Exception {
        namingService.deregisterInstance(RpcServiceHelper.buildServiceKeyWith_(serviceMeta.getServiceName(),
                serviceMeta.getServiceVersion()), serviceMeta.getServiceAddr(), serviceMeta.getServicePort());
        serviceNames.remove(serviceMeta.getServiceName());
    }

    @Override
    public ServiceMeta discovery(String serviceName, int invokerHashCode) throws Exception {
        if (serviceName.contains("#")) {
            serviceName = serviceName.replace("#", "_");
        }
        // todo
        List<Instance> allInstances = namingService.getAllInstances(serviceName);

        return null;
    }

    @Override
    public ServiceMeta discovery(String serviceName, String loadBalancer, int invokerHashCode) throws Exception {
        if (serviceName.contains("#")) {
            serviceName = serviceName.replace("#", "_");
        }
        List<Instance> allInstances = namingService.getAllInstances(serviceName);
        log.info("allInstances: {}", allInstances);
        List<ServiceMeta> serviceMetaList = allInstances.stream().map((item) -> {
            try {
                return mapper.readValue(item.getMetadata().get("serviceMeta"), ServiceMeta.class);
            } catch (JsonProcessingException e) {
                throw new NacosSerializationException();
            }
        }).collect(Collectors.toList());

        // 负载均衡算法
        ServiceLoadBalancer<ServiceMeta> balancer = LoadBalancerFactory
                .getLoadBalancer(LoadBalancerType.valueOf(loadBalancer));
        return balancer.select(serviceMetaList, invokerHashCode);
    }

    @Override
    public void destroy() throws IOException {
        if (!serviceNames.isEmpty() && address != null) {
            String host = address.getHostName();
            int port = address.getPort();
            Iterator<String> iterator = serviceNames.iterator();
            while (iterator.hasNext()) {
                String serviceName = iterator.next();
                try {
                    namingService.deregisterInstance(serviceName, host, port);
                } catch (NacosException e) {
                    throw new RuntimeException("注销服务失败");
                }
            }
        }
    }

    /**
     * 构建注册服务的实例对象
     *
     * @param serviceMeta 服务元数据信息
     * @return 实例对象
     */
    private Instance builderInstance(ServiceMeta serviceMeta) throws JsonProcessingException {
        Instance instance = new Instance();
        instance.setIp(serviceMeta.getServiceAddr());
        instance.setPort(serviceMeta.getServicePort());
        instance.setHealthy(false);
        Map<String, String> metaData = new HashMap<>(16);
        metaData.put("serviceMeta", new ObjectMapper().writeValueAsString(serviceMeta));
        instance.setMetadata(metaData);

        return instance;
    }
}
